package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @ClassName: Flink02_Table_Source_File
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 13:59
 *
 * 流程
 *      1、根据流环境，创建表环境
 *      2、创建表的模式，确定各个字段的模式Schema
 *      3、连接一个表,new FileSystem,创建一个临时表
 *          --确定表的字段的名称，类型
 *          --确定字段之间的分隔符
 *          --确定表的名称
 *      4、通过临时表获取一张表
 *      5、对表进行操作
 **/

public class Flink02_Table_Source_File {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        Schema schema = new Schema()
                            .field("id", DataTypes.STRING())
                            .field("ts",DataTypes.BIGINT())
                            .field("vc",DataTypes.INT());

        //连接，获取一个临时表
        tenv.connect(new FileSystem().path("in/sensor.txt"))
                //表的格式
                .withSchema(schema)
                //如何读取文件
                .withFormat(new Csv().fieldDelimiter(','))  //默认是，分割，行与行之间是\n分割
                //创建一个临时表
                .createTemporaryTable("senion");

        //通过表名获取一个表
        Table table = tenv.from("senion");

             table.select($("id"),$("ts"),$("vc"))
                  .execute().print();
    }



}
